package dataStream;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;


/**
 * @desc flink 读取 kafka 数据流 的基本操作
 * @Author xuyouchang
 * @Date 2021/8/5
 * @Update 2021/8/5 10:35
 */

public class KafkaSourceBasic {

    public static final String BOOTSTRAPS = "172.17.1.101:9092,172.17.1.102:9092,172.17.1.103:9092";

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(BOOTSTRAPS)
                .setTopics("after-vehicle-test")
                .setGroupId("my-group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .build();

        env.fromSource(
                source,
                WatermarkStrategy
                        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((record, ts) -> Long.parseLong(JSONObject.parseObject(record).getString("pass_time"))),

                "kafka source").print();
        env.execute();
    }
}
